package com.atguigu.flink.timer;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/10
 *
 *  定时器： 在指定的时间运行的 器物(程序)
 *
 *  ①制定定时器
 *          在什么时间触发运行，例如在3：00运行，需要使用时钟服务
 *          是什么时间语义， 如何知道当前已经3:00,看系统时间，还是看水印
 *  ②删除定时器
 *          在固定时间，没有达到之前，都可以删除定时器。到点后，指定的程序就不会触发
 *
 *  ③运行程序
 *          到点，就运行程序。仅仅运行一次。
 *
 *      --------------
 *          无keyBy，不定时
 *      UnsupportedOperationException: Setting timers is only supported on a keyed streams.
 *
 *      定时器是和key绑定的！
 *          每一种key各有各自的定时器，删除时，只能删除当前key的定时器。
 */
public class Demo1_ProcessingTimer
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KeyedStream<WaterSensor, String> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId);


        ds.process(new KeyedProcessFunction<String, WaterSensor, String>()
        {
            //编写定时器执行的代码
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                out.collect(ctx.getCurrentKey() + "在 "+timestamp +" 的定时器触发了....");

            }

            private long time ;

            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                //传感器的vc >= 10 ,在当前数据到达后的10s后，触发定时器, 如果vc < 10 ，就删除之前制定的定时器
                    TimerService timerService = ctx.timerService();

                    //单位是毫秒时间戳
                    if (value.getVc() >= 10){

                        time = timerService.currentProcessingTime() + 10000;
                        System.out.println(ctx.getCurrentKey() + "在" + time +" 注册了定时器...");
                        timerService.registerProcessingTimeTimer(time);
                    }else {
                        timerService.deleteProcessingTimeTimer(time);
                        System.out.println(ctx.getCurrentKey() + "在" + time +" 删除了定时器...");
                    }

                    out.collect(value.toString());

                }

        })
          .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
